package io.github.fb513.transport.queue;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import io.github.fb513.transport.config.BootstrapConfig;
import io.github.fb513.transport.config.RabbitMQConfig;
import io.github.fb513.transport.exception.InitServiceException;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author fengbo
 * @since 2023/12/25 15:58
 */
@Slf4j
public class MessageQueueSender {

    private final ThreadLocal<MessageChannel> channelThreadLocal;
    private final ThreadPoolExecutor executor;
    private final Connection connection;

    public MessageQueueSender(BootstrapConfig bootstrapConfig, ConnectionFactory connectionFactory) {
        channelThreadLocal = new ThreadLocal<>();
        RabbitMQConfig rabbitmqConfig = bootstrapConfig.getRabbitmq();
        int maxPoolSize = 100;
        if (rabbitmqConfig.getSenderChannelSize() != null) {
            maxPoolSize = rabbitmqConfig.getSenderChannelSize();
        }
        executor = new ThreadPoolExecutor(
                0, maxPoolSize,
                1L, TimeUnit.HOURS,
                new SynchronousQueue<>(),
                Thread.ofVirtual().name("im-transport-mq-sender-", 0).factory(),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        try {
            connection = connectionFactory.newConnection();
        } catch (IOException | TimeoutException e) {
            throw new InitServiceException("mq connect error", e);
        }
    }

    public void sendAsync(String exchangeName, String msg) throws RuntimeException {
        executor.execute(() -> {
            MessageChannel channel = channelThreadLocal.get();
            if (channel == null) {
                try {
                    channel = new MessageChannel(channelThreadLocal, connection.createChannel());
                } catch (IOException e) {
                    log.error("mq sender create channel error", e);
                    return;
                }
            }
            channelThreadLocal.set(channel);
            channel.publishMessage(exchangeName, msg);
        });
    }

    private static class MessageChannel {
        private final ThreadLocal<MessageChannel> threadLocal;
        private final Channel channel;
        private int failedNum;

        private MessageChannel(ThreadLocal<MessageChannel> threadLocal, Channel channel) {
            this.threadLocal = threadLocal;
            this.channel = channel;
            this.failedNum = 0;
        }

        private void publishMessage(String exchangeName, String msg) {
            try {
                channel.basicPublish(exchangeName, exchangeName, null, msg.getBytes());
                failedNum = 0;
            } catch (IOException e) {
                if (failedNum > 3) {
                    try {
                        channel.close();
                    } catch (IOException | TimeoutException ex) {
                        // ignored
                    }
                    threadLocal.remove();
                } else {
                    failedNum++;
                }
            }
        }
    }
}
